#include "ul_timer.h"
#include "ul_log_adapter.h"
#include "ul_malloc.h"
#include "ul_util.h"
#include <stdlib.h>
#include <gtest/gtest.h>
#include <iostream>
#include <vector>
#include "ul_pipe_task_queue.h"
using namespace LinuxNetworkUtility;
using namespace testing;
using namespace std;
char g_time_buf[sizeof(UlTimer)];
UlTimer *g_time = NULL;

char g_memory_pool_buf[sizeof(UlMemoryPool)];
UlMemoryPool *g_memory_pool = NULL;
UlModSet  g_mod_set;

void  __attribute__((constructor)) init_global_memory_pool()
{
  g_time = new(g_time_buf)UlTimer();
  EXPECT_EQ(g_time->init(10000),0);
  g_memory_pool = new(g_memory_pool_buf)UlMemoryPool(*g_time);
}

void  __attribute__((destructor)) destroy_global_memory_pool()
{
  g_memory_pool->~UlMemoryPool();
  g_memory_pool  = NULL;
  g_time->~UlTimer();
  g_time = NULL;
}

struct Tsk
{
  int64_t size_;
  char    buf_[0];
};

int32_t task_num = 0;
void *producer(void * arg)
{
  UlPipeMultiProducerSingleConsumerTQ *tsk_queue =
  reinterpret_cast<UlPipeMultiProducerSingleConsumerTQ*>(arg);
  for (int32_t i = 0; i < task_num; i++)
  {
    int64_t allocated_size = 0;
    static int64_t max_tsk_size = (2<<20);
    UNUSED(max_tsk_size);
    long randint = random();
    long tsk_size = randint%max_tsk_size + 1;
    Tsk *task = (Tsk*) ul_malloc(tsk_size + sizeof(Tsk),0,&allocated_size, FREE_TO_ALLOCATED_THREAD_POOL);
    EXPECT_GE(allocated_size, (int64_t)(tsk_size + sizeof(Tsk)));
    /// Tsk *task = (Tsk*)malloc(tsk_size + sizeof(Tsk));
    /// 
    EXPECT_NE(task, (void*)0);
    if (NULL != task)
    {
      task->size_ = tsk_size;
      ///memset(task->buf_, 0x00, task->size_);
      EXPECT_EQ(tsk_queue->push(task,0),0);
    }
  } 
  return NULL;
}

void usage(char **argv)
{
  fprintf(stderr, "%s <#producer> <#task_each_producer>\n", argv[0]);
  exit(0);
}

int main(int argc, char **argv)
{
  if (argc != 3)
  {
    usage(argv);
  }
  int32_t producer_num = atoi(argv[1]);
  task_num = atoi(argv[2]);
  if (producer_num <= 0 || task_num <= 0)
  {
    usage(argv);
  }

  EXPECT_EQ(ul_init_malloc(*g_memory_pool,*g_time,g_mod_set,10000),0);  
  pthread_t threads[producer_num];  

  UlPipeMultiProducerSingleConsumerTQ tsk_queue;
  EXPECT_EQ(tsk_queue.init(producer_num),0);

  /// create producer
  for (int32_t i = 0; i < producer_num; i++)
  {
    pthread_create(threads+i, NULL, producer, &tsk_queue);
  }
  /// consumer
  int64_t begin = ul_gettimeofday();
  const int32_t tsk_num = 8192;
  int32_t got_tsk_num = 0;
  int64_t total_got_tsk_num = 0;
  void *tsks[tsk_num];
  while (total_got_tsk_num < 1ll*producer_num*task_num)
  {
    got_tsk_num = tsk_queue.pop(tsks, tsk_num, 10000);
    if (got_tsk_num < 0)
    {
      EXPECT_TRUE(errno == EINTR || errno == EAGAIN);
    }
    else
    {
      UL_LOG_DEBUG("got tsk num:%d", got_tsk_num);
      total_got_tsk_num += got_tsk_num;
      for (int32_t i = 0; i < got_tsk_num; i++)
      {
        ul_free(tsks[i]);
        /// free(tsks[i]);
        tsks[i] = NULL;
      }
    }
  }
  int64_t used_time = ul_gettimeofday() - begin;
  fprintf(stderr, "used_time:%ld\n", used_time);

  /// join producer
  for (int32_t i = 0; i < producer_num; i++)
  {
    pthread_join(threads[i], NULL);
  }
}
